package com.young.study.mq.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class MyProducer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        //指定kafka服务器地址 如果是集群可以指定多个 但是就算只指定一个他也会去集群环境下寻找其他的节点地址
        properties.put("bootstrap.servers", "192.168.199.201:9092");

         /*ack是判别请求是否为完整的条件（就是是判断是不是成功发送了）。
         我们指定了“all”将会阻塞消息，表示所有的服务器发出ack 表示接受到消息之后这个消息才算发送成功，这种设置性能最低，但是是最可靠的,
         如果设置为整数，则表示当多少个服务器成功接受后就算发送成功*/

        //properties.put("ack", "all");
        //如果请求失败，生产者会自动重试，我们指定是0次，如果启用重试，则会有重复消息的可能性。
        //properties.put("retries", 0);
        //properties.put("batch.size", 16384);
        //在不处于高负载的情况下，如果设置比0大，以少量的延迟代价换取更少的，更有效的请求。
        //properties.put("linger.ms", 1);
        //控制生产者可用的缓存总量，如果消息发送速度比其传输到服务器的快，将会耗尽这个缓存空间。当缓存空间耗尽，其他发送调用将被阻塞，阻塞时间的阈值通过max.block.ms设定，之后它将抛出一个TimeoutException
        //properties.put("buffer.memory", 33554432);
        //将用户提供的key和value对象ProducerRecord转换成字节，你可以使用附带的ByteArraySerializaer或StringSerializer处理简单的string或byte类型

        properties.put("key.serializer", StringSerializer.class.getName());
        properties.put("value.serializer", StringSerializer.class.getName());

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        ProducerRecord<String,String> producerRecord = new ProducerRecord<>("my-topic",0,"test-key","hello");
        for (int i = 0; i < 10; i++) {
            System.out.println("消息发送："+i);
            //send()方法是异步的，调用该方法将会把消息添加消息到缓冲区等待发送，并立即返回
            Future<RecordMetadata> send = kafkaProducer.send(producerRecord);
            RecordMetadata recordMetadata = send.get();
            System.out.println(recordMetadata);

        }

        kafkaProducer.close();

    }


}
